package com.darcytech.debezium.core.api.bo;

import com.darcytech.debezium.core.api.RequestHandler;
import com.darcytech.debezium.core.pool.ConnectorInfo;
import com.google.common.base.Joiner;
import org.apache.commons.lang3.tuple.Pair;

import java.util.*;
import java.util.stream.Collectors;

public class RegisterInfo extends CCBindingHolder implements RequestHandler {

    public static final String SCHEMA_SUFFIX = "_cdc_ddl";

    /**
     * group为业务组，如:质检
     */
    String group;

    /**
     * 所有节点信息
     */
    Set<NodeInfo> nodeInfos;

    /**
     * 节点要监听的所有的表
     */
    Set<String> tables;

    public String getGroup() {
        return group;
    }

    public Set<NodeInfo> getNodeInfos() {
        return nodeInfos;
    }

    public Set<String> getTables() {
        return tables;
    }

    @Override
    public Object handle() {
        String group = getGroup();
        Set<String> tables = getTables();
        Set<RegisterInfo.NodeInfo> nodeInfos = getNodeInfos();

        Map<Pair<String, Integer>, Set<NodeInfo>> hostNameNodeInfos =
                nodeInfos.stream().collect(Collectors.groupingBy(node -> Pair.of(node.getHostName(), node.getPort()), Collectors.toSet()));

        List<ConnectorInfo> connectorInfoList = new ArrayList<>();

        for (Map.Entry<Pair<String, Integer>, Set<RegisterInfo.NodeInfo>> entry : hostNameNodeInfos.entrySet()) {
            Pair<String, Integer> hostAndPort = entry.getKey();
            Set<RegisterInfo.NodeInfo> nodes = entry.getValue();
            Set<String> dataBaseSet = new HashSet<>();
            Set<String> dataBaseTableSet = new HashSet<>();
            Integer port = null;
            String user = null;
            String password = null;
            for (RegisterInfo.NodeInfo node : nodes) {
                dataBaseSet.add(node.getDataBase());
                for (String table : tables) {
                    dataBaseTableSet.add(node.getDataBase() + "." + table);
                }
                user = node.getUser();
                password = node.getPassword();
                port = node.getPort();
            }

            String dataBases = Joiner.on(",").join(dataBaseSet);
            String dataBaseTables = Joiner.on(",").join(dataBaseTableSet);
            String hostName = hostAndPort.getLeft();
            String groupSchemaTopic = hostName + RegisterInfo.SCHEMA_SUFFIX;
            ConnectorInfo connectorInfo = new ConnectorInfo(group, hostName, port, dataBaseTables, user, password,
                    dataBases, groupSchemaTopic);
            connectorInfoList.add(connectorInfo);
        }

        List<String> warns = new ArrayList<>();
        for (ConnectorInfo connectorInfo : connectorInfoList) {
            try {
                if (!getConnectorConnectionBinding().registerInfo(connectorInfo)) {
                    warns.add("host:" + connectorInfo.getHostName() + ",port:" + connectorInfo.getPort() + " register error: the cluster server is switching");
                }
            } catch (Throwable t) {
                warns.add("host:" + connectorInfo.getHostName() + ",port:" + connectorInfo.getPort() + " register error: " + t);
            }
        }
        String message = Joiner.on("\n").join(warns);

        if (warns.size() == 0) {
            return new RegisterResult(RegisterResult.SUCCESS_CODE, "");
        }
        return new RegisterResult(RegisterResult.WARN_CODE, message);
    }

    public static class NodeInfo {

        String hostName;

        /**
         * dataBase 唯一
         */
        String dataBase;

        Integer port;

        String user;

        String password;

        public String getHostName() {
            return hostName;
        }

        public String getDataBase() {
            return dataBase;
        }

        public Integer getPort() {
            return port;
        }

        public String getUser() {
            return user;
        }

        public void setUser(String user) {
            this.user = user;
        }

        public String getPassword() {
            return password;
        }

        public void setPassword(String password) {
            this.password = password;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o)
                return true;
            if (o == null || getClass() != o.getClass())
                return false;
            NodeInfo nodeInfo = (NodeInfo) o;
            return Objects.equals(dataBase, nodeInfo.dataBase);
        }

        @Override
        public int hashCode() {
            return Objects.hash(dataBase);
        }
    }
}


